package yunjiao.javatutorials.guava.concurrent;

import com.google.common.util.concurrent.AbstractExecutionThreadService;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 消息处理服务
 *
 * @author yangyunjiao
 */
public class MessageProcessingService extends AbstractExecutionThreadService {

    private final BlockingQueue<Message> messageQueue;
    private final int batchSize;
    private int processedCount = 0;

    public MessageProcessingService(int queueCapacity, int batchSize) {
        this.messageQueue = new LinkedBlockingQueue<>(queueCapacity);
        this.batchSize = batchSize;
    }

    @Override
    protected void startUp() throws Exception {
        System.out.println("消息处理服务启动，批量大小: " + batchSize);
    }

    @Override
    protected void run() throws Exception {
        while (isRunning()) {
            try {
                Message message = messageQueue.take(); // 阻塞直到有消息
                processMessage(message);
                processedCount++;

                // 批量处理逻辑
                if (processedCount % batchSize == 0) {
                    System.out.println("已处理 " + processedCount + " 条消息");
                }

            } catch (InterruptedException e) {
                System.err.println(e.getMessage());
                if (!isRunning()) {
                    // 处理队列中剩余的消息
                    processRemainingMessages();
                    break;
                }
            }
        }
    }

    @Override
    protected void triggerShutdown() {
        System.out.println("消息处理服务停止信号接收，处理剩余消息...");
        processRemainingMessages();
    }

    @Override
    protected void shutDown() throws Exception {
        System.out.println("消息处理服务关闭，总共处理: " + processedCount + " 条消息");
    }

    private void processMessage(Message message) {
        // 模拟消息处理
        System.out.println("处理消息: " + message.content());
        try {
            Thread.sleep(600); // 模拟处理时间
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void processRemainingMessages() {
        Message message;
        while ((message = messageQueue.poll()) != null) {
            processMessage(message);
            processedCount++;
        }
    }

    public boolean offerMessage(Message message) {
        return messageQueue.offer(message);
    }

    public int getQueueSize() {
        return messageQueue.size();
    }

    // 消息类
    public record Message(String content, long timestamp) {
        public Message(String content) {
            this(content, System.currentTimeMillis());
        }
    }

    public static void main(String[] args) throws InterruptedException, TimeoutException {
        MessageProcessingService service = new MessageProcessingService(100, 10);

        // 1. 启动服务（异步）
        service.startAsync();
        service.awaitRunning(); // 阻塞等待服务启动完成

        for (int i = 0; i < 15; i++) {
            service.offerMessage(
                    new MessageProcessingService.Message("测试消息 " + i)
            );
        }
        System.out.printf("[主线程] 服务已启动（状态：%s）%n", service.state());

        // 2. 运行3秒后停止服务
        TimeUnit.SECONDS.sleep(3);
        service.stopAsync(); // 异步停止
        System.out.printf("[主线程] 已触发服务停止（状态：%s）%n", service.state());

        // 3. 等待服务终止（最多等待2秒）
        service.awaitTerminated(1, TimeUnit.SECONDS);
        System.out.printf("[主线程] 服务正常终止（状态：%s）%n", service.state());

    }
}
